{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "a22f15c6",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Python3_Jupyter_Nb_SparkSQL_讯捷集团门店QOQ-YOY分析.ipynb\n",
    "# Create By GF 2023-11-27 16:54"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "a7d07347",
   "metadata": {},
   "outputs": [],
   "source": [
    "import pyspark\n",
    "from pyspark import SparkContext, SparkConf\n",
    "from pyspark.sql import SparkSession, Row\n",
    "from pyspark.sql.types import IntegerType, DateType\n",
    "from pyspark.sql.window import Window\n",
    "from pyspark.sql.functions import asc, avg, col, desc, lead, lit, row_number, count, when"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "id": "441f9e77",
   "metadata": {},
   "outputs": [],
   "source": [
    "Prev_EMA_Value:float = 0.0\n",
    "    \n",
    "def BizInd_EMA(Period:int, Linenum:int, Target:float) -> float:\n",
    "\n",
    "    global Prev_EMA_Value\n",
    "    # --------------------------------------------------\n",
    "    if Linenum == 1:\n",
    "        Prev_EMA_Value = Target\n",
    "        # ----------------------------------------------\n",
    "        return round(Target, 4)\n",
    "    else:\n",
    "        Curr_EMA_Value = 2/(Period+1) * Target + (Period+1-2)/(Period+1) * Prev_EMA_Value\n",
    "        # ----------------------------------------------\n",
    "        Prev_EMA_Value = Curr_EMA_Value\n",
    "        # ----------------------------------------------\n",
    "        return round(Curr_EMA_Value, 4)\n",
    "\n",
    "def ObjRow_Add(ObjRow:pyspark.sql.types.Row, FldName:str, FldVal:object) -> pyspark.sql.types.Row: \n",
    "\n",
    "    Row_Dict = ObjRow.asDict() # -> Convert ObjRow to Dict. \n",
    "    # --------------------------------------------------\n",
    "    Row_Dict[FldName] = FldVal # -> Add a New Key in the Dictionary With the New Column Name and Value.\n",
    "    # --------------------------------------------------\n",
    "    New_ObjRow = Row(**Row_Dict) # -> Convert Dict to ObjRow. \n",
    "    # --------------------------------------------------\n",
    "    return New_ObjRow # -> Return New ObjRow."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "id": "da4c6e85",
   "metadata": {},
   "outputs": [],
   "source": [
    "# PySpark 创建 Spark 会话(连接)。\n",
    "# --------------------------------------------------\n",
    "#MySpark   = SparkSession.builder.appName('MySQL').getOrCreate()\n",
    "MyConf    = SparkConf().setMaster(\"local[*]\").setAppName(\"MySQL\")\n",
    "MySession = SparkSession.builder.config(conf=MyConf)\n",
    "SC        = SparkContext(conf=MyConf)\n",
    "MySpark   = MySession.getOrCreate()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "id": "8565d2fb",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "\n",
       "            <div>\n",
       "                <p><b>SparkSession - in-memory</b></p>\n",
       "                \n",
       "        <div>\n",
       "            <p><b>SparkContext</b></p>\n",
       "\n",
       "            <p><a href=\"http://xj:4040\">Spark UI</a></p>\n",
       "\n",
       "            <dl>\n",
       "              <dt>Version</dt>\n",
       "                <dd><code>v3.0.3</code></dd>\n",
       "              <dt>Master</dt>\n",
       "                <dd><code>local[*]</code></dd>\n",
       "              <dt>AppName</dt>\n",
       "                <dd><code>MySQL</code></dd>\n",
       "            </dl>\n",
       "        </div>\n",
       "        \n",
       "            </div>\n",
       "        "
      ],
      "text/plain": [
       "<pyspark.sql.session.SparkSession at 0x219c58e4430>"
      ]
     },
     "execution_count": 5,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "MySession.getOrCreate()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "id": "ffe7ead4",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[0, 0, 0, 0]"
      ]
     },
     "execution_count": 6,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "MyBroadcast = SC.broadcast([0,0,0,0])\n",
    "MyBroadcast.value"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "id": "ea966716",
   "metadata": {},
   "outputs": [],
   "source": [
    "Url        = \"jdbc:mysql://localhost:3306/gf_documents?useUnicode=ture&characterEncoding=UTF-8&useSSL=false\"\n",
    "Table_Name = \"view_xunjie_store_gross_profit_fill_project\"\n",
    "Properties = {\"user\": \"goufeng\", \"password\": \"12345678\"}"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "id": "9336635d",
   "metadata": {},
   "outputs": [],
   "source": [
    "df = MySpark.read.jdbc(url=Url, table=Table_Name, properties=Properties)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "id": "4eb7672f",
   "metadata": {},
   "outputs": [],
   "source": [
    "Rename_df = df\n",
    "# --------------------------------------------------\n",
    "Rename_df = Rename_df.withColumnRenamed(\"xj_date\",         \"日期\")\n",
    "Rename_df = Rename_df.withColumnRenamed(\"xj_year\",         \"年份\")\n",
    "Rename_df = Rename_df.withColumnRenamed(\"xj_month\",        \"月份\")\n",
    "Rename_df = Rename_df.withColumnRenamed(\"xj_ent_system\",   \"体系\")\n",
    "Rename_df = Rename_df.withColumnRenamed(\"xj_ent_depart\",   \"部门\")\n",
    "Rename_df = Rename_df.withColumnRenamed(\"xj_sto_name\",     \"门店名称\")\n",
    "Rename_df = Rename_df.withColumnRenamed(\"xj_sto_level\",    \"门店级别\")\n",
    "Rename_df = Rename_df.withColumnRenamed(\"xj_region_class\", \"区域类别\")\n",
    "Rename_df = Rename_df.withColumnRenamed(\"xj_gr_p\",         \"毛利\")\n",
    "Rename_df = Rename_df.withColumnRenamed(\"xj_gr_p_md_avg\",  \"毛利(月日均)\")\n",
    "Rename_df = Rename_df.withColumnRenamed(\"xj_gr_p_yd_avg\",  \"毛利(年日均)\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "id": "429b862a",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+----+----+----+----+--------------+--------+--------+---------+-------------+-------------+\n",
      "|      日期|年份|月份|体系|部门|      门店名称|门店级别|区域类别|     毛利| 毛利(月日均)| 毛利(年日均)|\n",
      "+----------+----+----+----+----+--------------+--------+--------+---------+-------------+-------------+\n",
      "|2021-09-29|2021|   9|形象|云南|云南丽江形象店|      B1|    二级|  3321.02|4645.57607143|3891.83511667|\n",
      "|2021-09-30|2021|   9|形象|云南|云南丽江形象店|      B1|    二级| 2492.438|4645.57607143|3891.83511667|\n",
      "|2021-09-24|2021|   9|形象|云南|云南丽江形象店|      B1|    二级|12184.018|4645.57607143|3891.83511667|\n",
      "|2021-09-25|2021|   9|形象|云南|云南丽江形象店|      B1|    二级|  11154.0|4645.57607143|3891.83511667|\n",
      "|2021-09-17|2021|   9|形象|云南|云南丽江形象店|      B1|    二级|  5156.69|4645.57607143|3891.83511667|\n",
      "|2021-09-18|2021|   9|形象|云南|云南丽江形象店|      B1|    二级|   5360.6|4645.57607143|3891.83511667|\n",
      "|2021-09-19|2021|   9|形象|云南|云南丽江形象店|      B1|    二级|   7700.8|4645.57607143|3891.83511667|\n",
      "|2021-09-22|2021|   9|形象|云南|云南丽江形象店|      B1|    二级| 3455.604|4645.57607143|3891.83511667|\n",
      "|2021-09-23|2021|   9|形象|云南|云南丽江形象店|      B1|    二级|  1866.74|4645.57607143|3891.83511667|\n",
      "|2021-09-26|2021|   9|形象|云南|云南丽江形象店|      B1|    二级| 4479.828|4645.57607143|3891.83511667|\n",
      "|2021-09-27|2021|   9|形象|云南|云南丽江形象店|      B1|    二级| 1215.114|4645.57607143|3891.83511667|\n",
      "|2021-09-28|2021|   9|形象|云南|云南丽江形象店|      B1|    二级| 13097.68|4645.57607143|3891.83511667|\n",
      "|2021-09-20|2021|   9|形象|云南|云南丽江形象店|      B1|    二级|  8523.92|4645.57607143|3891.83511667|\n",
      "|2021-09-21|2021|   9|形象|云南|云南丽江形象店|      B1|    二级|   5996.9|4645.57607143|3891.83511667|\n",
      "|2021-09-15|2021|   9|形象|云南|云南丽江形象店|      B1|    二级|   324.85|4645.57607143|3891.83511667|\n",
      "|2021-09-16|2021|   9|形象|云南|云南丽江形象店|      B1|    二级| 4165.988|4645.57607143|3891.83511667|\n",
      "|2021-09-08|2021|   9|形象|云南|云南丽江形象店|      B1|    二级|    848.7|4645.57607143|3891.83511667|\n",
      "|2021-09-09|2021|   9|形象|云南|云南丽江形象店|      B1|    二级|  5435.26|4645.57607143|3891.83511667|\n",
      "|2021-09-05|2021|   9|形象|云南|云南丽江形象店|      B1|    二级| 3684.275|4645.57607143|3891.83511667|\n",
      "|2021-09-06|2021|   9|形象|云南|云南丽江形象店|      B1|    二级|  1356.16|4645.57607143|3891.83511667|\n",
      "+----------+----+----+----+----+--------------+--------+--------+---------+-------------+-------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "ChangeType = Rename_df\n",
    "# --------------------------------------------------\n",
    "ChangeType = ChangeType.withColumn(\"年份\", ChangeType[\"年份\"].cast(IntegerType()))\n",
    "ChangeType = ChangeType.withColumn(\"月份\", ChangeType[\"月份\"].cast(IntegerType()))\n",
    "# --------------------------------------------------\n",
    "ChangeType.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "id": "0f1b312a",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+----+----+----+--------------------------+--------------+\n",
      "|年份|月份|体系|部门|                  门店名称|  毛利(月日均)|\n",
      "+----+----+----+----+--------------------------+--------------+\n",
      "|2022|   1|形象|云南|            云南丽江形象店| 3474.29835484|\n",
      "|2021|  11|形象|云南|      云南大理泰安路形象店| 3958.52893333|\n",
      "|2022|   6|形象|云南|      云南昆明正义坊形象店|14198.49842333|\n",
      "|2023|   5|形象|云南|      云南昆明正义坊形象店|14127.32297097|\n",
      "|2020|  12|形象|云南|    云南昆明环城东路形象店| 9839.98848387|\n",
      "|2022|   5|形象|云南|        云南昆明白云形象店| 8222.65863226|\n",
      "|2022|  11|形象|云南|      云南昆明白龙路形象店| 6938.90230333|\n",
      "|2022|   9|形象|云南|            云南曲靖形象店| 8554.47178667|\n",
      "|2022|   6|形象|云南|        云南曲靖百大形象店|     7747.1037|\n",
      "|2021|   6|形象|云南|            云南玉溪形象店| 5985.60427667|\n",
      "|2023|   7|形象|一部|      四川内江公园路形象店| 7848.57041935|\n",
      "|2023|   4|形象|一部|      四川大邑东街形象一店|18238.08437333|\n",
      "|2020|  12|形象|四部|        四川成都光华形象店| 5807.33713548|\n",
      "|2022|   4|形象|一部|        四川成都紫荆形象店| 6861.19376667|\n",
      "|2021|   4|形象|二部|    四川成都蜀汉路形象二店|    3413.97426|\n",
      "|2021|   4|形象|二部|      四川成都蜀汉路形象店| 3639.52242333|\n",
      "|2021|   5|形象|二部|      四川成都蜀汉路形象店| 4483.01118065|\n",
      "|2023|   3|形象|三部|四川成都锦华万达广场形象店|16752.52042581|\n",
      "|2021|   8|华为|四部|      四川新都新中路华为店| 9210.57058065|\n",
      "|2023|   1|形象|三部|        四川簇桥锦西形象店| 7014.32545161|\n",
      "+----+----+----+----+--------------------------+--------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# 聚合数据: 按年月。\n",
    "# --------------------------------------------------\n",
    "Agg_By_Year_Month = ChangeType\n",
    "# --------------------------------------------------\n",
    "Agg_By_Year_Month = Agg_By_Year_Month.groupBy([\"年份\", \"月份\", \"体系\", \"部门\", \"门店名称\"]).max(\"毛利(月日均)\")\n",
    "Agg_By_Year_Month = Agg_By_Year_Month.withColumnRenamed(\"max(毛利(月日均))\",  \"毛利(月日均)\")\n",
    "# --------------------------------------------------\n",
    "Agg_By_Year_Month.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "id": "d029e851",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+----+----+----+----------------------+-------------+----------------+-------+------------------+---------+\n",
      "|年份|月份|体系|部门|              门店名称| 毛利(月日均)|毛利(月日均)上期|linenum|       最近3月平均|     EMA3|\n",
      "+----+----+----+----+----------------------+-------------+----------------+-------+------------------+---------+\n",
      "|2020|  11|形象|四部|四川龙泉十陵灵龙形象店|       1000.0|            null|      1|            1000.0|   1000.0|\n",
      "|2020|  12|形象|四部|四川龙泉十陵灵龙形象店|       1000.0|          1000.0|      2|            1000.0|   1000.0|\n",
      "|2021|   1|形象|四部|四川龙泉十陵灵龙形象店|3250.86580645|          1000.0|      3|1750.2886021499999|2125.4329|\n",
      "|2021|   2|形象|四部|四川龙泉十陵灵龙形象店|2847.47714286|   3250.86580645|      4|2366.1143164366663| 2486.455|\n",
      "|2021|   3|形象|四部|四川龙泉十陵灵龙形象店| 2008.9483871|   2847.47714286|      5|     2702.43044547|2247.7017|\n",
      "|2021|   4|形象|四部|四川龙泉十陵灵龙形象店|     2082.039|    2008.9483871|      6|2312.8215099866666|2164.8704|\n",
      "|2021|   5|形象|四部|四川龙泉十陵灵龙形象店|2470.56808387|        2082.039|      7|     2187.18515699|2317.7192|\n",
      "|2021|   6|形象|四部|四川龙泉十陵灵龙形象店|     2474.027|   2470.56808387|      8|     2342.21136129|2395.8731|\n",
      "|2021|   7|形象|四部|四川龙泉十陵灵龙形象店|3511.78633548|        2474.027|      9|     2818.79380645|2953.8297|\n",
      "|2021|   8|形象|四部|四川龙泉十陵灵龙形象店|3932.44467742|   3511.78633548|     10|      3306.0860043|3443.1372|\n",
      "|2021|   9|形象|四部|四川龙泉十陵灵龙形象店|3029.45983333|   3932.44467742|     11|3491.2302820766668|3236.2985|\n",
      "|2021|  10|形象|四部|四川龙泉十陵灵龙形象店|2917.19012903|   3029.45983333|     12| 3293.031546593333|3076.7443|\n",
      "|2021|  11|形象|四部|四川龙泉十陵灵龙形象店|2146.62744667|   2917.19012903|     13|2697.7591363433335|2611.6859|\n",
      "|2021|  12|形象|四部|四川龙泉十陵灵龙形象店|2620.88530968|   2146.62744667|     14|     2561.56762846|2616.2856|\n",
      "|2022|   1|形象|四部|四川龙泉十陵灵龙形象店|2620.25073226|   2620.88530968|     15| 2462.587829536667|2618.2682|\n",
      "|2022|   2|形象|四部|四川龙泉十陵灵龙形象店|1502.98201786|   2620.25073226|     16|2248.0393532666667|2060.6251|\n",
      "|2022|   3|形象|四部|四川龙泉十陵灵龙形象店|1533.96232353|   1502.98201786|     17|1885.7316912166668|1797.2937|\n",
      "|2020|  11|华为|三部|        四川华阳华为店|4865.38576333|            null|      1|     4865.38576333|4865.3858|\n",
      "|2020|  12|华为|三部|        四川华阳华为店|4444.52016452|   4865.38576333|      2|    4654.952963925| 4654.953|\n",
      "|2021|   1|华为|三部|        四川华阳华为店|6328.61680323|   4444.52016452|      3|     5212.84091036|5491.7849|\n",
      "+----+----+----+----+----------------------+-------------+----------------+-------+------------------+---------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# 计算字段: 毛利环比基数。\n",
    "# --------------------------------------------------\n",
    "Profit_Qoq = Agg_By_Year_Month\n",
    "# --------------------------------------------------\n",
    "Profit_Qoq = Profit_Qoq.withColumn(\"毛利(月日均)上期\", lead(col(\"毛利(月日均)\"), 1)\n",
    "                                   .over(Window.partitionBy([\"体系\", \"部门\", \"门店名称\"])\n",
    "                                         .orderBy([desc(\"年份\"), desc(\"月份\")])))\n",
    "# --------------------------------------------------\n",
    "Profit_Qoq = Profit_Qoq.withColumn(\"linenum\", row_number()\n",
    "                                   .over(Window.partitionBy([\"体系\", \"部门\", \"门店名称\"])\n",
    "                                         .orderBy([asc(\"年份\"), asc(\"月份\")])))\n",
    "# --------------------------------------------------\n",
    "Profit_Qoq = Profit_Qoq.withColumn(\"最近3月平均\", avg(col(\"毛利(月日均)\"))\n",
    "                                   .over(Window.partitionBy([\"体系\", \"部门\", \"门店名称\"])\n",
    "                                         .orderBy([asc(\"年份\"), asc(\"月份\")])\n",
    "                                         .rowsBetween(-2, 0)))\n",
    "# --------------------------------------------------\n",
    "Profit_Qoq = Profit_Qoq.rdd.map(lambda x: ObjRow_Add(x, \"EMA3\", BizInd_EMA(3, x[\"linenum\"], x[\"毛利(月日均)\"]))).toDF()\n",
    "# --------------------------------------------------\n",
    "Profit_Qoq.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "id": "04d55d73",
   "metadata": {},
   "outputs": [],
   "source": [
    "# 将 Dataframe 保存为单个 CSV 文件 (coalesce(1) 方法将 DataFrame 的分区数设置为 1)。\n",
    "Profit_Qoq.coalesce(1).write.mode('overwrite').option('header', 'true').csv(\"file:///C:\\\\Csv_From_SparkSQL_讯捷集团门店QOQ-YOY分析.csv\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "id": "ee12d598",
   "metadata": {},
   "outputs": [],
   "source": [
    "# 关闭 Spark 会话。\n",
    "MySpark.stop()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "74601689",
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.8.0"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}
